use std::sync::{Condvar, mpsc, Mutex};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::mpsc::SendError;
use std::time::Duration;

use anyhow::{anyhow, bail, Result};
use lazy_static::lazy_static;

use crate::thread_safe_ptr;
use crate::utils::ThreadSafePtr;

///创建一个多线程控制工具
pub struct SimpleCountDownLatch {
    latch: AtomicUsize,
    inner_lock: Option<Mutex<()>>,
    inner_cond: Option<Condvar>,
}

impl SimpleCountDownLatch {
    pub fn new(latch: usize) -> Self {
        Self {
            latch: AtomicUsize::new(latch),
            inner_cond: None,
            inner_lock: None,
        }
    }

    pub fn new_with_lock(latch: usize) -> Self {
        Self {
            latch: AtomicUsize::new(latch),
            inner_cond: Some(Condvar::new()),
            inner_lock: Some(Mutex::new(())),
        }
    }

    pub fn count_down(&mut self) {
        self.latch.fetch_sub(1, Ordering::SeqCst);
        if let Some(cond) = self.inner_cond.as_ref() {
            cond.notify_one();
        };
    }
    fn judge_zero_and(counter: &AtomicUsize, mut fail_func: impl FnMut() -> () + Send + Sync) {
        while counter.load(Ordering::Acquire) > 0 {
            fail_func();
        }
    }
    pub fn wait(&self) {
        if let Some(lock) = &self.inner_lock {
            thread_safe_ptr! {
                let guard = Some(lock.lock().unwrap());
            }
            SimpleCountDownLatch::judge_zero_and(&self.latch, || {
                guard.read_write_ops(|tmp| {
                    Some(self.inner_cond.as_ref().unwrap().wait(tmp).unwrap())
                });
            });
        } else {
            SimpleCountDownLatch::judge_zero_and(&self.latch, || {});
        }
    }
}


/// 利用Send Receive特性实现一个countDownLatch
/// 注意仅支持wait单处理
pub struct SendReceiveCountDownLatch {
    count: AtomicUsize,
    count_down_fn: Box<dyn Fn(&mut AtomicUsize) + 'static + Send + Sync>,
    sender: mpsc::Sender<ThreadSafePtr<dyn Fn(&mut AtomicUsize) + 'static + Send + Sync>>,
    rev: mpsc::Receiver<ThreadSafePtr<dyn Fn(&mut AtomicUsize) + 'static + Send + Sync>>,
}

impl SendReceiveCountDownLatch {
    pub fn new(count: usize) -> Self {
        let (sender, rev) = mpsc::channel();
        Self {
            count: AtomicUsize::from(count),
            count_down_fn: Box::new(|atomic| { atomic.fetch_sub(1, Ordering::SeqCst); }),
            sender,
            rev,
        }
    }

    pub fn count_down(&mut self) -> Result<()> {
        Ok(self.sender.send(ThreadSafePtr::capture(&self.count_down_fn))?)
    }

    pub fn wait(&mut self) -> Result<()> {
        loop {
            self.rev.recv()?(&mut self.count);
            if self.count.load(Ordering::Acquire) == 0 {
                return Ok(());
            }
        }
    }

}


#[test]
fn test_count_down_latch() {
    thread_safe_ptr! {
       let latch = SendReceiveCountDownLatch::new(1);
    }
    (0..1).for_each(move |_| {
        std::thread::spawn(move || {
            std::thread::sleep(Duration::from_secs(5));
            latch.count_down().unwrap();
            let id = std::thread::current().id();
            println!("now thread {id:?} finished!!!");
        });
    });

    latch.wait().unwrap();
    println!("wait for all latch down");
}

#[test]
fn test_send_latch() {}